package com.spx.sql.window;

import com.spx.chapter05.pojo.Event;
import com.spx.util.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.e;

/**
 * create by undeRdoG on  2022-05-10  16:53
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class SetTimeTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 在创建表的DDL中定义时间属性
        String createDDL = "CREATE TABLE clickTable (" +
                "user_name STRING ," +
                "url STRING ," +
                "ts BIGINT," +
                "et AS TO_TIMESTAMP(FROM_UNIXTIME(ts/1000))," +
                "WATERMARK FOR et AS et - INTERVAL '1' SECONDS" +
                ") WITH (" +
                "'connector' = 'filesystem' ," +
                "'path' = 'input/clicks.txt' , " +
                "'format' = 'csv' "+
                ")";

        tableEnv.executeSql(createDDL);


        // 在流转换成Table的时候定义时间属性
        /*SingleOutputStreamOperator<Event> eventStream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );

        Table clickTable = tableEnv.fromDataStream(eventStream,
                $("user"),
                $("url"),
                $("timestamp").as("ts"),
                //rowtime() 单独定义一个事件时间下的时间语义字段
                $("timestamp").rowtime());*/

        //clickTable.printSchema();

        // 分组聚合
        Table aggTable = tableEnv.sqlQuery("select user_name,count(1) from clickTable group by user_name");
        tableEnv.toChangelogStream(aggTable).print("agg");

        env.execute();

    }

}
